package com.example.demo.producer;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

/**
 * <p> 消息发送者步骤分析
 * 1.创建消息生产者producer，并制定生产者组名
 * 2.指定Nameserver地址
 * 3.启动producer
 * 4.创建消息对象，指定主题Topic、Tag和消息体
 * 5.发送消息
 * 6.关闭生产者producer
 * </p>
 * @author Jon
 * @since 2021/7/13 16:34
 */
public class Producer {

    /*
     * @description 1.发送同步消息
     * @param       [args]
     * @return      void
     * @author      Jon
     * @since       2021/7/13 14:15
     */
//    public static void main( String[] args ) throws     Exception {
//        DefaultMQProducer producer = new DefaultMQProducer("syncProducer");
//        producer.setNamesrvAddr("127.0.0.1:9876");
//        producer.start();
//        producer.setSendMsgTimeout(30000);
//        for (int i = 0; i < 50000000; i++) {
//            // 创建消息，并指定Topic，Tag和消息体
//            Message msg = new Message("sendTopicSync" ,("Hello_RocketMQ " + i).getBytes("UTF-8"));
//            // 发送消息到一个Broker
//            SendResult sendResult = producer.send(msg);
//            // 通过sendResult返回消息是否成功送达
//            System.out.printf("%s%n", sendResult);
//            Thread.sleep(3000);
//        }
//        System.out.println("生产者发送了");
//        // 如果不再发送消息，关闭Producer实例
//        producer.shutdown();
//    }

   /*
    * @description 2.发送异步消息
    * @param       [args]
    * @return      void
    * @author      Jon
    * @since       2021/7/13 14:16
    */
    public static void main(String[] args) throws Exception {
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("asyncProducer");
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.start();
        //异步消息发送失败重试次数
        producer.setRetryTimesWhenSendAsyncFailed(5);

        for (int i = 0; i < 5; i++) {
            String topic = "topic_cib_hr";
            String tag = "topic_cib_hr_corp";
            String key = "order_id"+i;
            byte[] body = ("Hello Async_Msg" + i).getBytes();
            Message message = new Message(topic, tag, key, body);
            // 2.2 异步发送消息
            // SendCallback接收异步返回结果的回调
            producer.send(message, new SendCallback() {
                //rabbitmq急速入门的实战: 可靠性消息投递
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.err.println("msgId: " + sendResult.getMsgId() + ", status: " + sendResult.getSendStatus());
                }
                @Override
                public void onException(Throwable e) {
                    e.printStackTrace();
                    System.err.println("------发送失败");
                }
            });
        }
        //在producer.shutdown();前添加线程睡眠适当时间不然在异步发送消息之前，Producer就被关闭了，发送同步消息不会出现这种问题
        Thread.sleep(3000);
        // 如果不再发送消息，关闭Producer实例
        producer.shutdown();
    }

    /*
     * @description 3.单向发送消息 要用在不特别关心发送结果的场景，例如日志发送。
     * @param       [args]
     * @return      void
     * @author      Jon
     * @since       2021/7/13 14:20
     */
//    public static void main(String[] args) throws Exception {
//        DefaultMQProducer producer = new DefaultMQProducer("onewayProducer");
//        producer.setNamesrvAddr("127.0.0.1:9876");
//        producer.start();
//        Message message = new Message("TopicOneway","hello".getBytes(RemotingHelper.DEFAULT_CHARSET));
//        // 发送单向消息，没有任何返回结果
//        producer.sendOneway(message);
//        producer.shutdown();
//    }
}
